package com.webcat.redis.service;

import com.webcat.redis.exception.RedisRuntimeException;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisStringCommands;
import org.springframework.data.redis.connection.ReturnType;
import org.springframework.data.redis.core.*;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.data.redis.core.types.Expiration;

import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @author jun.wang
 * @title: RedisService
 * @projectName ownerpro
 * @description: TODO
 * @date 2019/5/7 10:05
 */

public class RedisService {

    private final RedisTemplate<String, Object> redisTemplate;

    public RedisService(RedisTemplate<String, Object> redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    public static final String UNLOCK_LUA;

    public static final String ADD_FIXED_RECORD_SCRIPT;

    static {
        UNLOCK_LUA = "if redis.call(\"get\",KEYS[1]) == ARGV[1] " +
                "then " +
                "    return redis.call(\"del\",KEYS[1]) " +
                "else " +
                "    return 0 " +
                "end ";

        ADD_FIXED_RECORD_SCRIPT = """
                local key = KEYS[1]
                local value = ARGV[1]
                local max_length = ARGV[2]
                redis.call("RPUSH", key, value)
                redis.call("LTRIM", key, 0, tonumber(max_length) - 1)
                return "OK"\s""";
    }

    /**
     * 设置缓存有效期
     * @param key
     * @param time
     * @return
     */
    public void expire(String key, long time) {
        redisTemplate.expire(key, time, TimeUnit.SECONDS);
    }

    /**
     * 根据key 获取过期时间
     * @param key 键 不能为null
     * @return 时间(秒) 返回0代表为永久有效
     */
    public long getExpire(String key) {
        Long val = redisTemplate.getExpire(key, TimeUnit.SECONDS);
        return val == null? 0:val;
    }

    /**
     * 判断key的存在性
     * @param key
     * @return
     */
    public boolean hasKey(String key) {
        Boolean b = redisTemplate.hasKey(key);
        return b != null && b;
    }
    /**
     * 删除缓存
     * @param key
     */
    public void delete(String ...key) {
        if (null != key && key.length > 0) {
            if (key.length == 1) {
                redisTemplate.delete(key[0]);
            }
            else {
                redisTemplate.delete(Arrays.asList(key));
            }
        }
    }


    /**
     * 获取缓存
     * @param key
     * @return
     */
    public<T> T get(String key) {
        return (T)redisTemplate.opsForValue().get(key);
    }

    /**
     * 设置缓存
     * @param key
     * @param value
     * @return
     */
    public void set(String key, Object value) {
        redisTemplate.opsForValue().set(key, value);
    }

    public boolean expireAt(String key, Date date) {
        Boolean b = redisTemplate.expireAt(key, date);
        return b != null && b;
    }

    /**
     * 设置缓存同时设置有效期
     * @param key
     * @param value
     * @param time
     * @return
     */
    public void set(String key, Object value, long time) {
        redisTemplate.opsForValue().set(key, value, time, TimeUnit.SECONDS);
    }

    /**
     * 递增
     * @param key
     * @param delta 递增步长
     * @return
     */
    public long incr(String key, long delta) {
        if (delta <= 0) {
            throw new RedisRuntimeException("delta must be greater than zero");
        }
        Long val = redisTemplate.opsForValue().increment(key, delta);
        return val == null? 0:val;
    }

    /**
     * 递减
     * @param key
     * @param delta 递减步长
     * @return
     */
    public long decr(String key, long delta) {
        if (delta <= 0) {
            throw new RedisRuntimeException("delta must be greater than zero");
        }
        Long val = redisTemplate.opsForValue().increment(key, -delta);
        return val == null? 0:val;
    }

    /**
     * 获取map中的某个item数据
     * @param key
     * @param item
     * @return
     */
    public<T> T hget(String key, String item) {
        if (null == item || item.isEmpty()) {
            throw new RedisRuntimeException("param item is empty");
        }
        return (T)redisTemplate.opsForHash().get(key, item);
    }

    /**
     * 设置map中的某个item数据
     * @param key
     * @param item
     * @param value
     * @return
     */
    public void hset(String key, String item, Object value) {
        if (null == item || item.isEmpty()) {
            throw new RedisRuntimeException("param item is empty");
        }
        redisTemplate.opsForHash().put(key, item, value);
    }

    /**
     * 获取map
     * @param key
     * @return
     */
    public<k, v> Map<k, v>hmget(String key) {
        return (Map<k, v>)redisTemplate.opsForHash().entries(key);
    }

    /**
     * 设置map
     * @param key
     * @param value
     * @return
     */
    public void hmset(String key, Map<?, ?> value) {
        redisTemplate.opsForHash().putAll(key, value);
    }

    /**
     * 删除hash表中的值
     * @param key
     * @param item
     * @return
     */
    public void hdel(String key, Object ...item) {
        redisTemplate.opsForHash().delete(key, item);
    }

    /**
     * 判断hash表中item是否存在
     * @param key
     * @param item
     * @return
     */
    public boolean hHasKey(String key, Object item) {
        return redisTemplate.opsForHash().hasKey(key, item);
    }


    /**
     * 获取分布式锁
     * @param key
     * @param value
     * @param time 单位秒
     * @return
     */
    public boolean setLock(String key, String value, long time) {
        RedisCallback<Boolean> callback = (connection) -> connection.set(key.getBytes(StandardCharsets.UTF_8), value.getBytes(StandardCharsets.UTF_8), Expiration.seconds(time), RedisStringCommands.SetOption.SET_IF_ABSENT);
        Boolean res = redisTemplate.execute(callback);
        return res != null && res;
    }

    /**
     * 释放分布式锁
     * @param key
     * @param value
     * @return
     */
    public boolean releaseLock(String key,String value) {
        RedisCallback<Boolean> callback = (connection) -> connection.eval(UNLOCK_LUA.getBytes(), ReturnType.BOOLEAN, 1, key.getBytes(StandardCharsets.UTF_8), value.getBytes(StandardCharsets.UTF_8));
        Boolean res = redisTemplate.execute(callback);
        return res != null && res;
    }

    /**
     * 存储固定数量的记录
     * @param key
     * @param value
     * @param maxSize
     * @return
     */
    public void addFixedRecord(String key, Object value, int maxSize) {
        DefaultRedisScript<Boolean> script = new DefaultRedisScript<>(ADD_FIXED_RECORD_SCRIPT, Boolean.class);
        redisTemplate.execute(script, Collections.singletonList(key), value, maxSize);
    }

    public void publish(String channel, Object msg) {
        redisTemplate.convertAndSend(channel, msg);
    }

    /**
     * zset新增单个元素
     * @param key
     * @param member
     * @param score
     */
    public void zAdd(String key, Object member, double score) {
        redisTemplate.opsForZSet().add(key, member, score);
    }

    /**
     * zset新增单个元素
     * @param key
     * @param member
     * @param score
     * @param timeout 过期时间,单位秒
     */
    public void zAdd(String key, Object member, double score, long timeout) {
        redisTemplate.opsForZSet().add(key, member, score);
        redisTemplate.expire(key, timeout, TimeUnit.SECONDS);
    }

    /**
     * zset新增多个元素
     * @param key
     * @param map
     */
    public void zAdd(String key, Map<Object, Double> map) {
        if (map != null && !map.isEmpty()) {
            Set<Object> keySet = map.keySet();
            Set<ZSetOperations.TypedTuple<Object>> memSet = keySet.stream().map(o -> {
                double score = map.get(o);
                return new DefaultTypedTuple<>(o, score);
            }).collect(Collectors.toSet());
            redisTemplate.opsForZSet().add(key, memSet);
        }
    }

    /**
     * zset新增多个元素
     * @param key
     * @param map
     * @param timeout 过期时间,单位秒
     */
    public void zAdd(String key, Map<Object, Double> map, long timeout) {
        zAdd(key, map);
        redisTemplate.expire(key, timeout, TimeUnit.SECONDS);
    }

    /**
     * zset根据索引查找元素
     * @param key
     * @param startIndex
     * @param endIndex
     * @return
     */
    public Set<Object> zRange(String key, long startIndex, long endIndex) {
        return redisTemplate.opsForZSet().range(key, startIndex, endIndex);
    }

    /**
     * zset根据索引查找元素与score
     * @param key
     * @param startIndex
     * @param endIndex
     * @return
     */
    public Set<ZSetOperations.TypedTuple<Object>> zRangeWithScores(String key, long startIndex, long endIndex) {
        return redisTemplate.opsForZSet().rangeWithScores(key, startIndex, endIndex);
    }

    /**
     * 增加或减少元素的score
     * @param key
     * @param member
     * @param score
     * @return
     */
    public Double zIncrBy(String key, Object member, double score) {
        return redisTemplate.opsForZSet().incrementScore(key, member, score);
    }

    /**
     * 获取元素的score
     * @param key
     * @param member
     * @return
     */
    public Double zScore(String key, Object member) {
        return redisTemplate.opsForZSet().score(key, member);
    }

    public List<Object> getList(String key, long start, long end) {
        return redisTemplate.opsForList().range(key, start, end);
    }

    public List<Object> getAllList(String key, long start) {
        return redisTemplate.opsForList().range(key, start, -1);
    }


    /**
     * key扫描
     * @param patternKey 匹配的keu表达式
     * @param count 扫描数量
     * @return key列表
     */
    public List<String> scan(String patternKey, int count) {
        ScanOptions options = ScanOptions.scanOptions().count(count).match(patternKey).build();
        Cursor<String> cursor = redisTemplate.scan(options);
        List<String> keyList = new ArrayList<>();
        while (cursor.hasNext()) {
            keyList.add(cursor.next());
        }
        return keyList;
    }

    /**
     * key扫描
     * @param patternKey
     * @return key列表
     */
    public List<String> scan(String patternKey) {
        return this.scan(patternKey, 1024);
    }

    /**
     * 批查询处理
     * @param keys
     * @return 批处理结果
     */
    public List<Object> get(String... keys) {
        return redisTemplate.executePipelined(new SessionCallback<>() {
            @Override
            public <K, V> Object execute(RedisOperations<K, V> operations) throws DataAccessException {
                Arrays.stream(keys).forEach(key -> operations.opsForValue().get(key));
                return null;
            }
        });
    }
}
